package org.example.aicloud.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import org.example.aicloud.util.AppVariable;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private ObjectMapper objectMapper;
    private static final String TOPIC = "ai-cloud-topic";
    @Resource
    private RedisTemplate redisTemplate;

    private static final String CANAL_TOPIC = "aicloud-canal-to-kafka";

    @Value("${mytest:www.cn.com}")
    private String mytest;

    @RequestMapping("/getconfig")
    public String getConfig() {
        return mytest;
    }
    @RequestMapping("/send")
    public String send(String msg) {
       kafkaTemplate.send(TOPIC,msg);
      return "ok";
    }

    @KafkaListener(topics = TOPIC)
    public void listen(String msg, Acknowledgment acknowledgment) {
        System.out.println("接受到的消息：" + msg);

        //手动确认消息
        acknowledgment.acknowledge();
    }

    @XxlJob("testJob")
    public void testJob(){
        System.out.println("执行 xxl job");
    }

    @RequestMapping("/test")
    public  String test() {
        return "test";
    }

    @KafkaListener(topics = CANAL_TOPIC)
    public void canalListen(String data, Acknowledgment acknowledgment) throws JsonProcessingException {
        HashMap<String, Object> map = objectMapper.readValue(data, HashMap.class);
        if (!map.isEmpty() && map.get("database").toString().equals("aicloud") &&
                map.get("table").toString().equals("answer")) {
            // 更新 Redis 缓存
            ArrayList<LinkedHashMap<String, Object>> list =
                    (ArrayList<LinkedHashMap<String, Object>>) map.get("data");
            String cacheKey = "";
            for (LinkedHashMap<String, Object> answer : list) {
                cacheKey = AppVariable.getListCacheKey(
                        Long.parseLong(answer.get("uid").toString()),
                        Integer.parseInt(answer.get("model").toString()),
                        Integer.parseInt(answer.get("type").toString()));

                redisTemplate.opsForValue().set(cacheKey, null); //该操作就是将我们有修改的数据从缓存中删除，这样可以确保下次读取是从数据库中读取到的。
            }
        }
        // 手动确认应答
        acknowledgment.acknowledge();

    }

}
